package com.htht.job.executor.activemq;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.TextMessage;

import com.htht.job.core.enums.MonitorQueue;
import com.htht.job.core.handler.SharingHandler;
import com.htht.job.executor.model.registry.Registry;
import com.htht.job.executor.service.registry.RegistryService;
import org.dom4j.Document;
import org.dom4j.DocumentException;
import org.dom4j.DocumentHelper;
import org.dom4j.io.OutputFormat;
import org.dom4j.io.XMLWriter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import com.alibaba.druid.util.StringUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.htht.job.core.biz.model.HandleCallbackParam;
import com.htht.job.core.biz.model.ReturnT;
import com.htht.job.core.biz.model.TriggerParam;
import com.htht.job.core.executor.XxlJobExecutor;
import com.htht.job.core.handler.IJobHandler;
import com.htht.job.core.log.XxlJobFileAppender;
import com.htht.job.core.log.XxlJobLogger;
import com.htht.job.core.thread.TriggerCallbackThread;
import com.htht.job.core.util.ScriptUtil;
import com.htht.job.executor.model.parameter.CommonParameter;
import com.htht.job.executor.redis.RedisService;
import com.htht.job.executor.util.SpringContextUtil;

@Component
public class QueueMessageListener implements MessageListener {
	public void onMessage(Message message) {  
		TriggerParam triggerParam = null;
        ReturnT<String> executeResult = null;
		RedisService redisService =null;
		RegistryService registryService = null;
		Registry registry=null;
		String key=null;
		try {
			redisService = (RedisService) SpringContextUtil.getBean("redisService");
			registryService = (RegistryService) SpringContextUtil.getBean("registryService");
			registry=registryService.findByMqDestination(registryService.getMqDestination());
			key=registry.getRegistryKey()+":"+registry.getRegistryIp()+registry.getPort();
			/**===========1.获取mq接受对象=========*/
			triggerParam = (TriggerParam)((ObjectMessage) message).getObject();
			//存储正在执行的job
			long nodeNum=redisService.getListSize(MonitorQueue.NODE_LINE_QUEUE +key);
			if (nodeNum>0l) {
				redisService.removeList(MonitorQueue.NODE_LINE_QUEUE + key, triggerParam.getJobId());
			}
			redisService.lPush(MonitorQueue.NODE_OPERATION_QUEUE+key, triggerParam.getJobId());
			redisService.lPush(  MonitorQueue.JOB_OPERATION_QUEUE+String.valueOf(triggerParam.getJobId()), triggerParam.getLogId());

			/**===========2.获取执行器=========*/
        	String handler=triggerParam.getExecutorHandler();
        	IJobHandler newJobHandler = XxlJobExecutor.loadJobHandler(handler);
        	/**===========3执行器执行业务代码=========*/
        	executeResult=newJobHandler.execute(triggerParam);
        	
			/**===========4.回调=========*/
            TriggerCallbackThread.pushCallBack(new HandleCallbackParam(triggerParam.getLogId(),executeResult,triggerParam.getLogFileName()));
			Thread.sleep(1000);
		} catch (Exception e) {
            ReturnT<String> stopResult = new ReturnT<String>(ReturnT.FAIL_CODE,   " 业务执行出错");
            TriggerCallbackThread.pushCallBack(new HandleCallbackParam(triggerParam.getLogId(), stopResult));
			 throw new RuntimeException();
		}finally {
			long jobNum=redisService.getListSize(MonitorQueue.JOB_OPERATION_QUEUE+String.valueOf(triggerParam.getJobId()));
			long nodeNum=redisService.getListSize(MonitorQueue.NODE_OPERATION_QUEUE+key);
			if(jobNum>0) {
				redisService.removeList(MonitorQueue.JOB_OPERATION_QUEUE + String.valueOf(triggerParam.getJobId()), triggerParam.getLogId());
			}
			if (nodeNum>0) {
				redisService.removeList(MonitorQueue.NODE_OPERATION_QUEUE + key, triggerParam.getJobId());
			}

		}
	}
  
}  